package top.jolyoulu.core.example

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Author: JolyouLu
 * @Date: 2024/2/6 16:32
 * @Description
 * 编号	字段名称	字段类型	字段含义
 * 1	date	String	用户点击行为的日期
 * 2	user_id	Long	用户的 ID
 * 3	session_id	String	Session 的 ID
 * 4	page_id	Long	某个页面的 ID
 * 5	action_time	String	动作的时间点
 * 6	search_keyword	String	用户搜索的关键词
 * 7	click_category_id	Long	某一个商品品类的 ID
 * 8	click_product_id	Long	某一个商品的 ID
 * 9	order_category_ids	String	一次订单中所有品类的 ID 集合
 * 10	order_product_ids	String	一次订单中所有商品的 ID 集合
 * 11	pay_category_ids	String	一次支付中所有品类的 ID 集合
 * 12	pay_product_ids	String	一次支付中所有商品的 ID 集合
 * 13	city_id	Long	城市 id
 */
object Spark01_Example_HotCategoryTop10PageflowAnalysis {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
    val sc: SparkContext = new SparkContext(sparkConf)

    //读取日志数据
    val path: String = this.getClass.getClassLoader.getResource("datas/user_visit_action.txt").toURI.getPath
    val actionRDD: RDD[String] = sc.textFile(path)
    val actionDataRDD: RDD[UserVisitAction] = actionRDD.map(
      action => {
        val datas: Array[String] = action.split("_")
        UserVisitAction(
          datas(0),
          datas(1).toLong,
          datas(2),
          datas(3).toLong,
          datas(4),
          datas(5),
          datas(6).toLong,
          datas(7).toLong,
          datas(8),
          datas(9),
          datas(10),
          datas(11),
          datas(12).toLong,
        )
      }
    )
    actionDataRDD.cache()
    //对指定的页面连续跳转进行统计
    //计算分母
    val ids = List(1L, 2L, 3L, 4L, 5L, 6L, 7L)
    val okFlowIds: List[(Long, Long)] = ids.zip(ids.tail)
    val pageidToCountMap: Map[Long, Long] = actionDataRDD.filter(
      action => {
        ids.init.contains(action.page_id)
      }
    ).map(
      action => {
        (action.page_id, 1L)
      }
    ).reduceByKey(_ + _).collect().toMap
    //计算分子
    //根据sessionId分组
    val sessionRDD: RDD[(String, Iterable[UserVisitAction])] = actionDataRDD.groupBy(_.session_id)
    //分组，排序
    val mvRDD: RDD[(String, List[((Long, Long), Int)])] = sessionRDD.mapValues(
      iter => {
        val sortList: List[UserVisitAction] = iter.toList.sortBy(_.action_time)
        val flowIds: List[Long] = sortList.map(_.page_id)
        val pageflowIds: List[(Long, Long)] = flowIds.zip(flowIds.tail)
        //将不合法的页面跳转过滤
        pageflowIds.filter(
          t => {
            okFlowIds.contains(t)
          }
        ).map(t => (t, 1))
      }
    )
    //((page1,page2),1)
    val flatRDD: RDD[((Long, Long), Int)] = mvRDD.map(_._2).flatMap(list => list)
    //((page1,page2),sum)
    val dataRDD: RDD[((Long, Long), Int)] = flatRDD.reduceByKey(_ + _)
    //计算单跳转换率
    dataRDD.foreach {
      case (((pageid1, pageid2), sum)) => {
        val lon: Long = pageidToCountMap.getOrElse(pageid1, 0L)
        println(s"页面${pageid1}跳转到页面${pageid2}单跳转化率为：" + (sum.toDouble / lon))
      }
    }
    sc.stop()
  }

  //用户访问动作表
  case class UserVisitAction(
                              date: String, //用户点击行为的日期
                              user_id: Long, // 用 户 的 ID
                              session_id: String, //Session 的 ID
                              page_id: Long, // 某 个 页 面 的 ID
                              action_time: String, //动作的时间点
                              search_keyword: String, //用户搜索的关键词
                              click_category_id: Long, // 某 一 个 商 品 品 类 的 ID
                              click_product_id: Long, // 某 一 个 商 品 的 ID
                              order_category_ids: String, //一次订单中所有品类的 ID 集合
                              order_product_ids: String, //一次订单中所有商品的 ID 集合
                              pay_category_ids: String, //一次支付中所有品类的 ID 集合
                              pay_product_ids: String, //一次支付中所有商品的 ID 集合
                              city_id: Long //城市 id
                            )

}